package org.baitu.zzim.service;

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.baitu.zzim.model.ChatMessage;
import org.baitu.zzim.producer.ChatMessageProducer;
import org.baitu.zzim.util.RedisUtil;
import org.springframework.stereotype.Service;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
 * @author zhangkai_a
 * @date 2025/7/2 09:34
 */
@Service
public class ChannelManager {
    private final RedisUtil redisUtil;
    private final ChatMessageProducer producer;
    private final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private final ConcurrentMap<String, Channel> userChannels = new ConcurrentHashMap<>();

    public ChannelManager(RedisUtil redisUtil, ChatMessageProducer producer) {
        this.redisUtil = redisUtil;
        this.producer = producer;
    }

    public void addChannel(String userId, Channel channel) {
        channels.add(channel);
        userChannels.put(userId, channel);
        redisUtil.saveUserNode(userId);
    }

    public void removeChannel(String userId, Channel channel) {
        channels.remove(channel);
        userChannels.entrySet().removeIf(entry -> entry.getValue().equals(channel));
        redisUtil.removeUserNode(userId);
    }

    public boolean containsChannel(String userId) {
        return userChannels.containsKey(userId);
    }
    /**
     * 发送消息给指定用户
     * 1. 先判断本地是否有该用户的连接，存在直接发送
     * 2. 不存在则查 Redis 获取该用户所在节点，并通过 RocketMQ 发送到对应节点
     */
    public void sendToUser(String userId, Object message) {
        Channel channel = userChannels.get(userId);
        if (channel != null && channel.isActive()) {
            // 本地用户，直接写
            channel.writeAndFlush(message);
        } else {
            // 不在本地，跨节点推送
            String nodeId = redisUtil.getUserNode(userId);
            if (nodeId == null) {
                // 用户离线或未注册
                System.out.println("用户 " + userId + " 不在线");
                return;
            }

            if (!(message instanceof ChatMessage)) {
                throw new IllegalArgumentException("消息必须是 ChatMessage 类型");
            }

            // 使用 RocketMQTemplate 顺序发送，key 用 userId 保证顺序
            producer.sendMessage(message,userId);
        }
    }

    public void broadcast(Object message) {
        channels.writeAndFlush(message);
    }
}